package com.alan.weather.service.weather;

import com.alan.weather.entity.weather.WeatherData_MouthOfYear;
import com.alan.weather.entity.weather.WeatherData_OneDay;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.util.Scanner;

public class Step3_8 {
    static WeatherData_MouthOfYear result = new WeatherData_MouthOfYear();
    private static class Step3_8Mappper extends Mapper<LongWritable, Text, Text, WeatherWritable> {
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, WeatherWritable>.Context context) throws IOException, InterruptedException {
            String line = value.toString();
            // 数据验证
            if (StringUtils.isBlank(line)){
                return;
            }
            // 防止数据拆分
            String[] items = line.split("\t");
            if (items == null || items.length != 5){
                return;
            }
            // 获取查询年份
            String year = context.getConfiguration().get("year");
            // 过滤年份
            if (items[0].indexOf(year) < 0){
                return;
            }
            // 实例化对象
            WeatherWritable w = new WeatherWritable(items[0],
                    Double.parseDouble(items[1]),
                    Double.parseDouble(items[2]),
                    Double.parseDouble(items[3]),
                    Double.parseDouble(items[4]));

            String code = items[0].split("_")[0];
            // 取月份
            String month = items[0].split("_")[1].substring(3,5);
            // 输出
            // context.write(new Text(code + "_" + year), w);
            context.write(new Text(code + "\t" + month), w);
        }
    }

    private static class Step3_8Reducer extends Reducer<Text, WeatherWritable,  WeatherWritable, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<WeatherWritable> values, Reducer<Text, WeatherWritable,  WeatherWritable, NullWritable>.Context context) throws IOException, InterruptedException {
            WeatherWritable w = new WeatherWritable(key.toString(), 0.0, 0.0, 0.0, 0.0);
            int flag = 0;
            for (WeatherWritable v : values){
                if (flag == 0){
                    w.setPrecipitation(v.getPrecipitation());
                    w.setMaxTemperature(v.getMaxTemperature());
                    w.setMinTemperature(v.getMinTemperature());
                    w.setAvgTemperature(v.getAvgTemperature());
                    flag++;
                }
                w.setPrecipitation((w.getPrecipitation() + v.getPrecipitation()) / 2);
                w.setMaxTemperature((w.getMaxTemperature() + v.getMaxTemperature()) / 2);
                w.setMinTemperature((w.getMinTemperature() + v.getMinTemperature()) / 2);
                w.setAvgTemperature((w.getAvgTemperature() + v.getAvgTemperature()) / 2);
            }
            // 输出
            context.write(w, NullWritable.get());

            WeatherData_OneDay temp = new WeatherData_OneDay();
            temp.date = w.getCode_date().split("\t")[1];
            temp.averageTemperature = w.getAvgTemperature().toString();
            temp.highestTemperature = w.getMaxTemperature().toString();
            temp.lowestTemperature = w.getMinTemperature().toString();
            temp.precipitation = w.getPrecipitation().toString();
            result.mouths.add(temp);
        }
    }

    public static WeatherData_MouthOfYear run(String input, String output, String year) {
        try {
            result.mouths.clear();
            // 传递日期:通过配置对象将参数设置为全局
            HadoopUtils.getConf().set("year",year);
            // 定义输入路径
            // String input = "/brazil_weather";
            // 定义输出路径
            // String output = "/step1_output";
            Path outputPath = new Path(output);
            // 判断输出目录是否存在，存在则删除之
            HadoopUtils.delete(HadoopUtils.getFileSystem(), outputPath, true);
            // 定义Job任务
            Job job = Job.getInstance(HadoopUtils.getConf(), "step3_8");
            //设置Jar包
            job.setJarByClass(Step1.class);
            // 设置输入
            job.setInputFormatClass(TextInputFormat.class);
            FileInputFormat.setInputPaths(job, input);
            // 设置Mapper
            job.setMapperClass(Step3_8Mappper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(WeatherWritable.class);
            // 设置Reducer
            job.setReducerClass(Step3_8Reducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            // 设置输出
            job.setOutputFormatClass(TextOutputFormat.class);
            FileOutputFormat.setOutputPath(job, outputPath);
            // 运行
            boolean flag = job.waitForCompletion(true);
            // 提示
            if (flag){
                System.out.println("Step3_8:查询某年每个月的天气数据~~~");
                // 显示数据
                return result;
            }
        }
        catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }
}
